#include "../header/ThreadPool.h"

ThreadPool::ThreadPool(int keep_alive_time, int core_thread_number, int maximum_thread_number, queue<Task> *task_queue,
                       ThreadFactory *factory) :
        keep_alive_time(keep_alive_time), core_thread_number(core_thread_number),
        maximum_thread_number(maximum_thread_number), task_queue(task_queue), factory(factory) {

    if (task_queue == nullptr || factory == nullptr) {
        printf("The Task Queue Or Thread Pool Factory Parameter Is Abnormal!\n");
        return;
    }

    // 线程池线程数量相关的参数
    work_thread_number = 0;
    alive_thread_number = 0;

    // 初始化线程池的条件变量和互斥锁
    pthread_mutex_init(&task_queue_mutex, nullptr);
    pthread_mutex_init(&thread_pool_mutex, nullptr);
    pthread_cond_init(&task_queue_is_fill_cond, nullptr);
    pthread_cond_init(&task_queue_is_empty_cond, nullptr);

    // 设置工作线程PID列表的容量
    work_thread_pid_list.resize(maximum_thread_number);

    // 创建管理者线程
    manager_thread_pid = factory->createThread(managerThreadWorkFunction, this);
}

ThreadPool::ThreadPool(int keep_alive_time, int core_thread_number, int maximum_thread_number, queue<Task> *task_queue,
                       ThreadFactory *factory, Time_Unit time_unit, RejectionPolicy policy) :
        keep_alive_time(keep_alive_time), core_thread_number(core_thread_number),
        maximum_thread_number(maximum_thread_number),
        task_queue(task_queue), factory(factory), time_unit(time_unit), policy(policy) {

    ThreadPool(keep_alive_time, core_thread_number, maximum_thread_number, task_queue, factory);
}

ThreadPool::~ThreadPool() {
    printf("Start Thread Pool Recovery Work!\n");

    shutdown();

    // 回收管理者线程
    pthread_join(manager_thread_pid, nullptr);

    // 回收工作线程
    int count = 0;
    bool isCompleteWork = false;
    while (!isCompleteWork && count++ != 100) {
        // 线程每间隔 300 毫秒就检测工作线程的状态
        timespec sleep_time = {0, 1000 * 1000 * 300};
        nanosleep(&sleep_time, nullptr);
        isCompleteWork = true;
        for (int i = 0; i < maximum_thread_number; ++i) {
            pthread_t pid = work_thread_pid_list[i];
            if (pid != 0) {
                isCompleteWork = false;
                pthread_cond_signal(&task_queue_is_empty_cond);
            }
        }
    }

    // 超过半分钟线程就让工作线程强制退出
    for (int i = 0; i < maximum_thread_number; ++i) {
        pthread_t pid = work_thread_pid_list[i];
        if (pid != 0) {
            pthread_cancel(pid);
        }
    }

    // 回收互斥锁和条件变量
    pthread_mutex_destroy(&task_queue_mutex);
    pthread_mutex_destroy(&thread_pool_mutex);
    pthread_cond_destroy(&task_queue_is_fill_cond);
    pthread_cond_destroy(&task_queue_is_empty_cond);

    // 回收堆资源
    delete factory;
    delete task_queue;

    // 回收服务器的文件描述符
    if (server_fd != -1) {
        close(server_fd);
    }
}

int ThreadPool::getNewWorkThreadIndex() {
    for (int i = 0; i < maximum_thread_number; ++i) {
        if (work_thread_pid_list[i] == 0) {
            return i;
        }
    }

    return -1;
}

void *ThreadPool::workThreadWorkFunction(void *args) {
    printf("Work Thread PID %ld Complete Create!\n", pthread_self());

    ThreadPool *pool = static_cast<ThreadPool *>(args);

    // 存放当前线程的存活时间
    pool->work_thread_alive_time_map.insert(make_pair(pthread_self(), time(nullptr) + pool->keep_alive_time));
    while (pool->state != Terminated) {
        usleep(1000 * (rand() % 500 + 200));
        pthread_mutex_lock(&pool->thread_pool_mutex);
        pthread_mutex_lock(&pool->task_queue_mutex);
        while (pool->task_queue->empty()) {
            printf("Task Queue Size Is Empty, Work Thread Block Wait Signal!\n");

            // 释放当前线程获取到的锁资源
            pthread_mutex_unlock(&pool->task_queue_mutex);
            pthread_cond_wait(&pool->task_queue_is_empty_cond, &pool->thread_pool_mutex);
            pthread_mutex_lock(&pool->task_queue_mutex);

            // 线程池为结束态时, 工作线程进行自我销毁
            // 线程池为关闭态时, 工作线程正常执行任务队列的任务, 但线程池不能提交任务
            if (pool->state == Terminated) {
                pthread_mutex_unlock(&pool->thread_pool_mutex);
                pthread_mutex_unlock(&pool->task_queue_mutex);
                workThreadExitFunction(pool);
            }

            // 查看当前线程是否到了存活时间
            pthread_t pid = pthread_self();
            long current_time = time(nullptr);
            long alive_time = pool->work_thread_alive_time_map[pid];
            // 线程池为关闭态时, 忽略工作线程的生存时间, 进行执行任务, 直到任务队列为空时才执行退出工作
            if (alive_time <= current_time && (pool->state != Shutdown || pool->task_queue->empty())) {
                pthread_mutex_unlock(&pool->thread_pool_mutex);
                pthread_mutex_unlock(&pool->task_queue_mutex);
                workThreadExitFunction(pool);
            }
        }
        pthread_mutex_unlock(&pool->thread_pool_mutex);

        // 从任务队列中获取任务
        Task task = pool->task_queue->front();
        pool->task_queue->pop();
        pthread_mutex_unlock(&pool->task_queue_mutex);

        // 通知阻塞的任务提交线程
        pthread_cond_signal(&pool->task_queue_is_fill_cond);

        // 工作线程执行任务
        pool->work_thread_number++;
        task.function(task.args);
        // 此处释放内存存在问题, 这样操作只能释放指针指向的内存, 成员变量的指向的内存是不会释放的
        delete task.args;
        pool->work_thread_number--;
        // 重置工作线程的销毁时间
        pool->work_thread_alive_time_map.insert(make_pair(pthread_self(), time(nullptr) + pool->keep_alive_time));
    }
    workThreadExitFunction(pool);

    return nullptr;
}

void *ThreadPool::managerThreadWorkFunction(void *args) {
    printf("Manager Thread PID %ld Complete Create!\n", pthread_self());

    ThreadPool *pool = static_cast<ThreadPool *>(args);

    while (pool->state == Running) {
        // 管理者线程每间隔 300 毫秒就进行线程池状态的检测
        timespec sleep_time = {0, 1000 * 1000 * 300};
        nanosleep(&sleep_time, nullptr);

        // 提示超出存活时间的线程进行自我销毁
        for (int i = 0; i < pool->maximum_thread_number; ++i) {
            pthread_mutex_lock(&pool->thread_pool_mutex);
            pthread_t pid = pool->work_thread_pid_list[i];
            if (pid != 0) {
                long alive_time = pool->work_thread_alive_time_map[pid];
                long current_time = time(nullptr);

                if (alive_time != 0 && alive_time <= current_time) {
                    pthread_cond_signal(&pool->task_queue_is_empty_cond);
                }
            }
            pthread_mutex_unlock(&pool->thread_pool_mutex);
        }

        // 任务队列存在任务时, 通知工作线程去处理
        if (!pool->task_queue->empty()) {
            pthread_cond_signal(&pool->task_queue_is_empty_cond);
        }
    }

    printf("Manager Thread PID %ld Complete Destroy!\n", pthread_self());

    return nullptr;
}

void ThreadPool::workThreadExitFunction(void *args) {
    ThreadPool *pool = static_cast<ThreadPool *>(args);
    pthread_t pid = pthread_self();

    pthread_mutex_lock(&pool->thread_pool_mutex);
    pool->work_thread_alive_time_map.erase(pid);
    for (int i = 0; i < pool->maximum_thread_number; ++i) {
        if (pool->work_thread_pid_list[i] == pid) {
            pool->work_thread_pid_list[i] = 0;
        }
    }
    pthread_mutex_unlock(&pool->thread_pool_mutex);

    pool->alive_thread_number--;
    printf("Work Thread PID %ld Complete Destroy!\n", pid);

    pthread_exit(nullptr);
}

void ThreadPool::shutdown() {
    state = Shutdown;
}

bool ThreadPool::submitTask(void *args, callback function) {
    return submitTask(Task(args, function));
}

bool ThreadPool::submitTask(Task task) {
    if (state == Running) {
        pthread_mutex_lock(&thread_pool_mutex);
        // 两种需要新建线程的情况:
        // 1.存活线程数等于工作线程数且存活线程数小于线程池核心线程数
        // 2.任务队列已满且存活线程小于线程最大线程数
        if ((alive_thread_number.getValue() == work_thread_number.getValue() &&
             alive_thread_number.getValue() < core_thread_number) ||
            (alive_thread_number.getValue() < maximum_thread_number && task_queue->size() == TASK_QUEUE_CAPACITY)) {
            printf("New Thread Needs To Be Created! Thread Number = %d\n", alive_thread_number.getValue());
            int index = getNewWorkThreadIndex();
            if (index != -1) {
                work_thread_pid_list[index] = factory->createThread(workThreadWorkFunction, this);
                alive_thread_number++;
            }
        }
        pthread_mutex_unlock(&thread_pool_mutex);

        bool isExecuteTask = false;
        pthread_mutex_lock(&task_queue_mutex);
        if (task_queue->size() < TASK_QUEUE_CAPACITY) {
            printf("Push Thread Pool Task!\n");
            task_queue->push(task);
            pthread_cond_signal(&task_queue_is_empty_cond);

        } else {
            // 根据线程池的拒绝策略执行对应的操作
            switch (policy) {
                case AbortPolicy:
                    printf("The Thread Pool Is Full, Throwing An Exception To The Upper Layer!\n");
                    break;

                case CallerRunsPolicy:
                    printf("The Thread Pool Is Full, Let The Task Committer Execute The Task!\n");
                    isExecuteTask = true;
                    break;

                case BlockingWaitPolicy:
                    while (true) {
                        if (task_queue->size() != TASK_QUEUE_CAPACITY) {
                            task_queue->push(task);
                        } else {
                            printf("The Thread Pool Is Full, Blocking Waiting For The Task To Be Committed!\n");
                            pthread_cond_wait(&task_queue_is_fill_cond, &thread_pool_mutex);
                        }
                        break;
                    }
                    break;

                case DiscardPolicy:
                    printf("The Thread Pool Is Full, Discard Current Task!\n");
                    // 此处释放内存存在问题, 这样操作只能释放指针指向的内存, 成员变量的指向的内存是不会释放的
                    delete task.args;
                    break;

                case DiscardOldestPolicy:
                    task_queue->pop();
                    task_queue->push(task);
                    printf("The Thread Pool Is Full. Discard The Pre-Task!\n");
                    break;

                default:
                    break;
            }
        }
        pthread_mutex_unlock(&task_queue_mutex);

        // 让提交任务的线程自己处理任务
        if (isExecuteTask) {
            task.function(task.args);
        }

        return true;
    }

    printf("The Thread Pool Has Been Closed And Cannot Continue To Submit Task!\n");
    return false;
}

int ThreadPool::getWorkThreadNumber() {
    return work_thread_number.getValue();
}

int ThreadPool::getAliveThreadNumber() {
    return alive_thread_number.getValue();
}